package com.twoandone.tms.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.twoandone.tms.realtime.common.TmsConfig;
import com.twoandone.tms.realtime.utils.DimUtil;
import com.twoandone.tms.realtime.utils.ThreadPoolUtil;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.util.Collections;
import java.util.concurrent.ExecutorService;

/**
 * @packageName:com.twoandone.tms.realtime.utils
 * @ClassName:DwsAsyncUtil
 * @Description:
 * @author:
 * @date 2023/3/11 16:48
 * 模板方法设计模式
 **/
public abstract class DwsAsyncUtil<T> extends RichAsyncFunction<T, T> implements DimJoinFunction<T> {
    private String table;
    private ExecutorService thread;

    public DwsAsyncUtil(String table) {
        this.table = table;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        thread = ThreadPoolUtil.getInstance();

    }

    /*
     * invoke 没来一条数据，就会执行一次，不能将thread放在这里实例化
     * 这就需要使用open方法，复函数才有这些冬习
     * 那就去 使用复函数！
     * */
    @Override
    public void asyncInvoke(T obj, ResultFuture<T> resultFuture) throws Exception {

        thread.submit(new Runnable() {
            @Override
            public void run() {
//      获取维度id关联字段
                String key = getKey(obj);
//                    用key 去维度拿数据
                JSONObject dim = DimUtil.getDimInfo(table.toUpperCase(), key);
//      将维度添加到 主流中
                join(obj, dim);
                resultFuture.complete(Collections.singleton(obj));
            }
        });
    }


}
